<?php
namespace Swoole\Registry;

use Swoole\Socket\IServer;
use Swoole\Socket\IClient;
use Swoole\Socket\SocketServer;
use Common\Log;
use Swoole\Socket\SocketClient;
use Swoole\Registry\RegistryClient;
/**
 * 注册中心服务器
 * 
 *
 */
class RegistryServer implements IServer{
    const THRIFT_SERVER = 1;
    const THRIFT_CLIENT = 2;
    const THRIFT_REGISTER = 3;
    private $config;
    private $port;
    private $host;
    private $registerServers = array();
    private $thriftServers = array();
    const TIMEOUT = 120; //sec
    function __construct($config, $port = 0){
        //print_r($config);
        $this->config = $config;
        
    }
    
    /**
     * 初始化所有注册中心地址，
     * 接收thrift server的服务器信息注册
     */
    public function start(){
        $address = $this->config['address'];
//         var_dump($address);
//         Log::info($address);
        $address = parse_url($address);
        if(empty($address) == true){
            exit('address is correct');
        }
        //         $keep_time = 30; // 存活时间
        $this->host = $address['host'];
        $this->port = $address['port'];
        $this->initRegisterServer();
        if(empty($address['host']) == true || empty($address['port']) == true){
            exit('error, host or port can not be empty');
        }

        
        $set = array(
            //    'worker_num' => 8,   //工作进程数量 PHP代码中是全异步非阻塞，worker_num配置为CPU核数的1-4倍即可。如果是同步阻塞，worker_num配置为100或者更高
            //    'daemonize' => true, //是否作为守护进程
            'open_eof_check' => true,
            'package_eof' => "\r\n"
        );
        $server = new SocketServer($this, $this->host, $this->port, $set);
        $server->start();
//         
//         $thrift_servers = array();
        
    }
    
    public function onStart($serv){
        echo "info: register server run at {$this->host}:{$this->port} \n";
    }
    
    public function onConnect($serv, $fd, $from_id){
        //echo __CLASS__ . " ====== onConnect \n";
        $fdinfo = $serv->connection_info($fd);
        $remote_ip = $fdinfo["remote_ip"];
        $remote_port = $fdinfo["remote_port"];
        echo "Client Connect, remote_ip:$remote_ip,remote_port:$remote_port.\n";
    }
    
    /**
     * 接收来自thrift server和registry server服务器信息注册,并把其它thrift server服务器信息返回给连接的server
     * 把当前所有thrift server服务器信息同步到其它注册中心服务器
     * 接收三种消息：1、thrift server信息更新，2、thrift client请求最新可用thrif server，3、其它registry server同步过来的信息
     */
    public function onReceive($serv, $fd, $from_id, $data){
//         echo __CLASS__ . " ====== onReceive \n";
        $fdinfo = $serv->connection_info($fd);
        $remote_ip = $fdinfo["remote_ip"];
        $remote_port = $fdinfo["remote_port"];
        
        echo "on receive1:".$data."\n";
        echo "on receive2:".json_encode($this->thriftServers)."\n";
        $data = json_decode($data, true);
        if(empty($data) == false){
            //$type = $data['type'];
            $this->handleReceive($serv, $fd, $data);
        }
        else{
            echo "on receive data format error\n";
        }
    }
    
    public function onClose($serv, $fd, $from_id){
        $fdinfo = $serv->connection_info($fd);
        $remote_ip = $fdinfo["remote_ip"];
        $remote_port = $fdinfo["remote_port"];
        $clentKey = $remote_ip.':'.$remote_port;
        unset($this->thrift_servers[$clentKey]);
        echo "Client {$remote_ip}:{$remote_port}: Close.==== \n";
      
    }
    /**
     * 
     * @param unknown $serv
     * @param unknown $fd
     * @param unknown $data {type:1,msg:{}}
     */
    public function handleReceive($serv, $fd, $data){
        $fdinfo = $serv->connection_info($fd);
        $remote_ip = $fdinfo["remote_ip"];
        $remote_port = $fdinfo["remote_port"];
        //         $clentKey = $remote_ip.':'.$remote_port;
        $type = $data['type'];
        switch($type){
            case RegistryServer::THRIFT_SERVER:
                $this->handleThriftServerMessage($serv, $fd, $data);
                $this->sendToRegistryServer($serv, $fd, $data);
                break;
            case RegistryServer::THRIFT_CLIENT:
                $this->handleThriftClientMessage($serv, $fd, $data);
                break;
            case RegistryServer::THRIFT_REGISTER:
                $this->handleRegisterServerMessage($serv, $fd, $data);
                break;
        }

    }
    
    /**
     * 处理来自thrift server的数据
     * @param unknown $serv
     * @param unknown $fd
     * @param unknown $data
     */
    private function handleThriftServerMessage($serv, $fd, $data){
        $fdinfo = $serv->connection_info($fd);
//         $remote_ip = $fdinfo["remote_ip"];
//         $remote_port = $fdinfo["remote_port"];
//         $clentKey = $remote_ip.':'.$remote_port;

        $clentKey = $data['host'].':'.$data['port'];
        $nowtime = time();
        $data['time'] = $nowtime;
        $this->thriftServers[$clentKey] = $data;
        $activeServers = array();
        foreach($this->thriftServers as $key => $server){
            // 如果超时不更新，则删除当前可用server
            if(($nowtime - $server['time']) < RegistryServer::TIMEOUT){
                $activeServers[] = array('host'=>$server['host'],'port'=>$server['port']);
            }
            else{
                unset($this->thriftServers[$key]);
            }
        }
        if(count($activeServers) > 0){
            $sendData = json_encode($activeServers);
            //echo "send to $clentKey:" . $sendData."\n";
            $serv->send($fd, $sendData."\r\n");
        }
        //$this->updateRegisterServer($data);
//         }
    }
    
    /**
     * 
     * @param unknown $serv
     * @param unknown $fd
     * @param unknown $data
     */
    private function sendToRegistryServer($serv, $fd, $data){
        $data['type'] = RegistryServer::THRIFT_REGISTER;
        foreach($this->registerServers as $key => $client){
            if($client->isConnected() == false){
                list($host, $port) = explode(':', $key);
                $port = intval($port);
                if($client->connect() == false){
                    echo "connect error $host:$port :Error: \n";
                }
                else{
                    $this->registerServers[$key] = $client;
                }
            }
            else{
                $data = json_encode($data);
//                 echo "s {$key}===send:".json_encode($data)."\n"; 
                Log::info("update to other register server ".json_encode($data));
                $client->send($data);
            }
        }
    }
    
    /**
     * 处理来自thrift client的请求信息
     * @param unknown $serv
     * @param unknown $fd
     * @param unknown $data
     */
    private function handleThriftClientMessage($serv, $fd, $data){
        $thriftServers = $this->thriftServers;
        if(count($thriftServers) > 0){
            $nowtime = time();
            $activeServers = array();
            foreach($this->thriftServers as $key => $server){
                // 如果超时不更新，则删除当前可用server
                if(($nowtime - $server['time']) < RegistryServer::TIMEOUT){
                    $activeServers[] = array('host'=>$server['host'],'port'=>$server['port']);
                }
                else{
                    unset($this->thriftServers[$key]);
                }
            }
            Log::info("handleThriftClientMessage===" . json_encode($activeServers));
            if(count($activeServers) > 0){
                $sendData = json_encode($activeServers);
                //echo "send to $clentKey:" . $sendData."\n";
                $serv->send($fd, $sendData."\r\n");
            }
        }
    }

    /**
     * 来自其它注册中心的同步消息
     * @param unknown $serv
     * @param unknown $fd
     * @param unknown $data
     */
    private function handleRegisterServerMessage($serv, $fd, $data){
        $clentKey = $data['host'].':'.$data['port'];
        $nowtime = time();
        $data['time'] = $nowtime;
        $this->thriftServers[$clentKey] = $data;
        //$activeServers = array();
        foreach($this->thriftServers as $key => $server){
            // 如果超时不更新，则删除当前可用server
            if(($nowtime - $server['time']) < RegistryServer::TIMEOUT){
                //$activeServers[] = array('host'=>$server['host'],'port'=>$server['port']);
            }
            else{
                unset($this->thriftServers[$key]);
            }
        }
    }

    private function initRegisterServer(){
        $servers = $this->config['registry'];
        if(count($servers) > 0){
            $localKey = $this->host.':'.$this->port;
            //echo '$localKey='.$localKey."\n";
            foreach($servers as $key=>$s){
                
                $s = parse_url($s);
//                 var_dump($s);
                $host = $s['host'];
                $port = $s['port'];
                $k = $host.":" . $port;
//                 echo 'key='.$k."\n";
                Log::info('registry server='.$k);
                if($localKey != $k){
                    Log::info('connect registry server='.$k);
                    $client = new RegistryClient($host, $port);
                    //$client->connect();
                    $this->registerServers[$k] = $client;
                    //echo "\nclient->isConnected()=" .$client->isConnected()."\n";
                }
            }
        }
    }
}


?>